package main

import (
	"encoding/json"
	"fmt"
	"io/ioutil"
	"log"
	"sync"
)

// Task 定义单个任务
type Task struct {
	Name  string `json:"name"`
	Type  string `json:"type"`
	Input int    `json:"input"`
}

// TaskResult 定义任务结果
type TaskResult struct {
	Name    string
	Status  string
	Message string
	Result  int
}

// Stage 定义一个阶段
type Stage struct {
	Name      string `json:"name"`
	Execution string `json:"execution"`
	Tasks     []Task `json:"tasks"`
}

// Pipeline 定义整个管道
type Pipeline struct {
	Stages []Stage `json:"stages"`
}

// Execute 执行管道
func (p *Pipeline) Execute() {
	for _, stage := range p.Stages {
		fmt.Printf("Executing Stage: %s\n", stage.Name)

		switch stage.Execution {
		case "parallel":
			p.executeParallel(stage.Tasks)
		case "serial":
			p.executeSerial(stage.Tasks)
		default:
			fmt.Printf("Unknown execution type: %s\n", stage.Execution)
		}
	}
}

// executeParallel 并行执行任务
func (p *Pipeline) executeParallel(tasks []Task) {
	var wg sync.WaitGroup
	results := make([]TaskResult, len(tasks))

	for i, task := range tasks {
		wg.Add(1)
		go func(i int, t Task) {
			defer wg.Done()
			result := executeTask(t)
			status := "success"
			message := "Task completed successfully"

			if result < 0 { // 假设小于0为失败
				status = "fail"
				message = "Task failed"
			}

			results[i] = TaskResult{
				Name:    t.Name,
				Status:  status,
				Message: message,
				Result:  result,
			}
		}(i, task)
	}
	wg.Wait() // 等待所有并行任务完成
	p.logResults(results)
}

// executeSerial 串行执行任务
func (p *Pipeline) executeSerial(tasks []Task) {
	results := make([]TaskResult, len(tasks))

	for i, task := range tasks {
		result := executeTask(task)
		status := "success"
		message := "Task completed successfully"

		if result < 0 { // 假设小于0为失败
			status = "fail"
			message = "Task failed"
		}

		results[i] = TaskResult{
			Name:    task.Name,
			Status:  status,
			Message: message,
			Result:  result,
		}
	}
	p.logResults(results)
}

// executeTask 模拟任务执行
func executeTask(t Task) int {
	if t.Type == "simple" {
		return t.Input * 2 // 简单任务示例
	}
	return -1 // 模拟失败的情况
}

// logResults 记录任务结果
func (p *Pipeline) logResults(results []TaskResult) {
	for _, result := range results {
		fmt.Printf("Task %s: Status - %s, Message - %s, Result - %d\n",
			result.Name, result.Status, result.Message, result.Result)
	}
}

// LoadPipeline 从 JSON 文件加载管道
func LoadPipeline(filename string) (*Pipeline, error) {
	data, err := ioutil.ReadFile(filename)
	if err != nil {
		return nil, err
	}

	var pipeline Pipeline
	err = json.Unmarshal(data, &pipeline)
	if err != nil {
		return nil, err
	}

	return &pipeline, nil
}

func main() {
	pipeline, err := LoadPipeline("pipeline.json")
	if err != nil {
		log.Fatalf("Error loading pipeline: %v", err)
	}

	pipeline.Execute()
}
